package com.rk.recommand;

import cn.hutool.db.nosql.redis.RedisDS;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import redis.clients.jedis.Jedis;

import java.io.IOException;

public class RedisOutputFormat<K, V> extends FileOutputFormat<K, V> {

    protected static class RedisRecordWriter<K, V> extends RecordWriter<K, V>{

        private Jedis jedis;

        public RedisRecordWriter(Jedis jedis){
            this.jedis = jedis;
        }

        public void write(K key, V value) throws IOException, InterruptedException {
            boolean nullkey = key == null;
            boolean nullvalue = value == null;
            if (nullkey || nullvalue) return;

            System.out.println("k: " + key + ": " + "v: " + value);
            jedis.del("recommand_key:" + key);
            jedis.append("recommand_key:" + key, value.toString());
        }

        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            if (jedis != null){
                jedis.disconnect();
            }
        }
    }


    public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        Jedis jedis = RedisDS.create().getJedis();
        return new RedisRecordWriter(jedis);
    }
}
